--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit f10a29d21a0da3ae8eea27ea62b985b9198d05fb
Parents : 8371613
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-03-06T12:03:50-06:00
Improve crash recovery and monitoring capabilities.
Changes
11 files changed, 896 insertions(+), 19 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index bdff6801..7e077155 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -88,7 +88,7 @@ from meshchatx.src.backend.nomadnet_utils import (
convert_nomadnet_string_data_to_map,
)
from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler
-from meshchatx.src.backend.recovery import CrashRecovery
+from meshchatx.src.backend.recovery import CrashRecovery, HealthMonitor
from meshchatx.src.backend.rnprobe_handler import RNProbeHandler
from meshchatx.src.backend.sideband_commands import SidebandCommands
from meshchatx.src.backend.telemetry_utils import Telemeter
@@ -744,6 +744,19 @@ class ReticulumMeshChat:
# Link database to memory log handler
memory_log_handler.set_database(context.database)
+ # Wire crash recovery with DB + log handler for adaptive diagnostics
+ if hasattr(self, "_crash_recovery") and self._crash_recovery:
+ self._crash_recovery.set_database(context.database)
+ self._crash_recovery.log_handler = memory_log_handler
+
+ # Start health monitor if not already running
+ if not hasattr(self, "_health_monitor") or self._health_monitor is None:
+ self._health_monitor = HealthMonitor(
+ log_handler=memory_log_handler,
+ app=self,
+ )
+ self._health_monitor.start()
+
def _checkpoint_and_close(self):
# delegated to database instance
self.database._checkpoint_and_close()
@@ -12419,6 +12432,9 @@ def main():
docs_download_urls=args.docs_download_urls,
)
+ # store recovery on app for wiring with identity context
+ reticulum_meshchat._crash_recovery = recovery
+
# update recovery with known paths
recovery.update_paths(
storage_dir=reticulum_meshchat.storage_dir,
diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py
index 5d37e947..2ce37bf4 100644
--- a/meshchatx/src/backend/database/__init__.py
+++ b/meshchatx/src/backend/database/__init__.py
@@ -8,6 +8,7 @@ from datetime import UTC, datetime
from .announces import AnnounceDAO
from .config import ConfigDAO
from .contacts import ContactsDAO
+from .crash_history import CrashHistoryDAO
from .debug_logs import DebugLogsDAO
from .legacy_migrator import LegacyMigrator
from .map_drawings import MapDrawingsDAO
@@ -41,6 +42,7 @@ class Database:
self.contacts = ContactsDAO(self.provider)
self.map_drawings = MapDrawingsDAO(self.provider)
self.debug_logs = DebugLogsDAO(self.provider)
+ self.crash_history = CrashHistoryDAO(self.provider)
def initialize(self):
self._tune_sqlite_pragmas()
diff --git a/meshchatx/src/backend/database/crash_history.py b/meshchatx/src/backend/database/crash_history.py
new file mode 100644
index 00000000..1bd5b661
--- /dev/null
+++ b/meshchatx/src/backend/database/crash_history.py
@@ -0,0 +1,65 @@
+import json
+
+from .provider import DatabaseProvider
+
+
+class CrashHistoryDAO:
+ def __init__(self, provider: DatabaseProvider):
+ self.provider = provider
+
+ def insert_crash(
+ self,
+ timestamp,
+ error_type,
+ error_message,
+ diagnosed_cause,
+ symptoms,
+ probability,
+ entropy,
+ divergence,
+ ):
+ self.provider.execute(
+ """
+ INSERT INTO crash_history
+ (timestamp, error_type, error_message, diagnosed_cause,
+ symptoms, probability, entropy, divergence)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ (
+ timestamp,
+ error_type,
+ error_message[:500] if error_message else None,
+ diagnosed_cause,
+ json.dumps(symptoms) if isinstance(symptoms, dict) else symptoms,
+ probability,
+ entropy,
+ divergence,
+ ),
+ )
+
+ def get_recent_crashes(self, limit=50):
+ return self.provider.fetchall(
+ "SELECT * FROM crash_history ORDER BY timestamp DESC LIMIT ?",
+ (limit,),
+ )
+
+ def get_cause_frequencies(self, limit=50):
+ return self.provider.fetchall(
+ """
+ SELECT diagnosed_cause, COUNT(*) as count
+ FROM (SELECT * FROM crash_history ORDER BY timestamp DESC LIMIT ?)
+ GROUP BY diagnosed_cause
+ ORDER BY count DESC
+ """,
+ (limit,),
+ )
+
+ def cleanup_old(self, max_entries=200):
+ self.provider.execute(
+ """
+ DELETE FROM crash_history WHERE id NOT IN (
+ SELECT id FROM crash_history ORDER BY timestamp DESC LIMIT ?
+ )
+ """,
+ (max_entries,),
+ )
diff --git a/meshchatx/src/backend/database/schema.py b/meshchatx/src/backend/database/schema.py
index eebca3f6..38e8bef4 100644
--- a/meshchatx/src/backend/database/schema.py
+++ b/meshchatx/src/backend/database/schema.py
@@ -13,7 +13,7 @@ def _validate_identifier(name: str, label: str = "identifier") -> str:
class DatabaseSchema:
- LATEST_VERSION = 39
+ LATEST_VERSION = 40
def __init__(self, provider: DatabaseProvider):
self.provider = provider
@@ -1046,6 +1046,25 @@ class DatabaseSchema:
"CREATE INDEX IF NOT EXISTS idx_lxmf_messages_state_peer ON lxmf_messages(state, peer_hash)",
)
+ if current_version < 40:
+ self._safe_execute("""
+ CREATE TABLE IF NOT EXISTS crash_history (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ timestamp REAL,
+ error_type TEXT,
+ error_message TEXT,
+ diagnosed_cause TEXT,
+ symptoms TEXT,
+ probability INTEGER,
+ entropy REAL,
+ divergence REAL,
+ created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+ )
+ """)
+ self._safe_execute(
+ "CREATE INDEX IF NOT EXISTS idx_crash_history_timestamp ON crash_history(timestamp)",
+ )
+
# Update version in config
self._safe_execute(
"""
diff --git a/meshchatx/src/backend/persistent_log_handler.py b/meshchatx/src/backend/persistent_log_handler.py
index 3dadde35..6260d637 100644
--- a/meshchatx/src/backend/persistent_log_handler.py
+++ b/meshchatx/src/backend/persistent_log_handler.py
@@ -1,10 +1,13 @@
import collections
import logging
+import math
import re
import threading
import time
from datetime import UTC, datetime
+_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
+
class PersistentLogHandler(logging.Handler):
def __init__(self, database=None, capacity=5000, flush_interval=5):
@@ -27,6 +30,10 @@ class PersistentLogHandler(logging.Handler):
self.known_ips = set()
self.known_uas = set()
+ # Entropy tracking — level counts over a sliding 60s window
+ self._level_events = collections.deque(maxlen=10000)
+ self._error_events = collections.deque(maxlen=10000)
+
def set_database(self, database):
with self.lock:
self.database = database
@@ -49,6 +56,10 @@ class PersistentLogHandler(logging.Handler):
with self.lock:
self.logs_buffer.append(log_entry)
+ now_mono = time.monotonic()
+ self._level_events.append((now_mono, record.levelname))
+ if record.levelno >= logging.ERROR:
+ self._error_events.append(now_mono)
# Periodically flush to database if available
if self.database and (
@@ -233,3 +244,34 @@ class PersistentLogHandler(logging.Handler):
is_anomaly=is_anomaly,
)
return len(self.logs_buffer)
+
+ @property
+ def current_log_entropy(self):
+ """Shannon entropy over log-level distribution in the last 60 seconds."""
+ cutoff = time.monotonic() - 60.0
+ with self.lock:
+ counts = {lv: 0 for lv in _LOG_LEVELS}
+ total = 0
+ for ts, level in self._level_events:
+ if ts >= cutoff:
+ counts[level] = counts.get(level, 0) + 1
+ total += 1
+ if total == 0:
+ return 0.0
+ entropy = 0.0
+ for c in counts.values():
+ if c > 0:
+ p = c / total
+ entropy -= p * math.log2(p)
+ return entropy
+
+ @property
+ def current_error_rate(self):
+ """Fraction of log events at ERROR or above in the last 60 seconds."""
+ cutoff = time.monotonic() - 60.0
+ with self.lock:
+ total = sum(1 for ts, _ in self._level_events if ts >= cutoff)
+ errors = sum(1 for ts in self._error_events if ts >= cutoff)
+ if total == 0:
+ return 0.0
+ return errors / total
diff --git a/meshchatx/src/backend/recovery/__init__.py b/meshchatx/src/backend/recovery/__init__.py
index 5992cca6..4a36ac2f 100644
--- a/meshchatx/src/backend/recovery/__init__.py
+++ b/meshchatx/src/backend/recovery/__init__.py
@@ -1,3 +1,4 @@
from .crash_recovery import CrashRecovery
+from .health_monitor import HealthMonitor
-__all__ = ["CrashRecovery"]
+__all__ = ["CrashRecovery", "HealthMonitor"]
diff --git a/meshchatx/src/backend/recovery/crash_recovery.py b/meshchatx/src/backend/recovery/crash_recovery.py
index 7cda9f8e..709c46ed 100644
--- a/meshchatx/src/backend/recovery/crash_recovery.py
+++ b/meshchatx/src/backend/recovery/crash_recovery.py
@@ -1,26 +1,45 @@
-"""CRASH RECOVERY & DIAGNOSTIC ENGINE
-------------------------------------------
-This module implements a mathematically grounded diagnostic system for MeshChatX.
-It utilizes Active Inference heuristics, Shannon Entropy, and KL-Divergence
+"""CRASH RECOVERY & ADAPTIVE DIAGNOSTIC ENGINE
+--------------------------------------------------
+Mathematically grounded diagnostic system for MeshChatX.
+
+Uses Shannon Entropy, KL-Divergence, and Bayesian weight learning
to map application failures onto deterministic manifold constraints.
+Crash history is persisted and priors are refined over time using
+a conjugate Beta-Binomial model.
"""
import contextlib
+import json
import os
import platform
import re
import shutil
import sqlite3
import sys
+import time
import traceback
import psutil
import RNS
+_DEFAULT_PRIORS = {
+ "DB_SYNC_FAILURE": 0.05,
+ "DB_CORRUPTION": 0.05,
+ "ASYNC_RACE": 0.10,
+ "OOM": 0.02,
+ "CONFIG_MISSING": 0.01,
+ "RNS_IDENTITY_FAILURE": 0.05,
+ "LXMF_STORAGE_FAILURE": 0.05,
+ "INTERFACE_OFFLINE": 0.05,
+ "UNSUPPORTED_PYTHON": 0.05,
+ "LEGACY_SYSTEM_LIMITATION": 0.05,
+}
+
class CrashRecovery:
"""A diagnostic utility that intercepts application crashes and provides
- meaningful error reports and system state analysis.
+ meaningful error reports and system state analysis. Learns from crash
+ history to refine root-cause probabilities over time.
"""
def __init__(
@@ -29,12 +48,17 @@ class CrashRecovery:
database_path=None,
public_dir=None,
reticulum_config_dir=None,
+ database=None,
+ log_handler=None,
):
self.storage_dir = storage_dir
self.database_path = database_path
self.public_dir = public_dir
self.reticulum_config_dir = reticulum_config_dir
+ self.database = database
+ self.log_handler = log_handler
self.enabled = True
+ self._learned_priors = None
# Check environment variable to allow disabling the recovery system
env_val = os.environ.get("MESHCHAT_NO_CRASH_RECOVERY", "").lower()
@@ -69,6 +93,95 @@ class CrashRecovery:
if reticulum_config_dir:
self.reticulum_config_dir = reticulum_config_dir
+ def set_database(self, database):
+ """Provide database access for crash persistence and weight learning."""
+ self.database = database
+ self._load_learned_priors()
+
+ def _load_learned_priors(self):
+ """Load learned Bayesian priors from the config table."""
+ if not self.database:
+ return
+ try:
+ raw = self.database.config.get("diagnostic_weights")
+ if raw:
+ self._learned_priors = json.loads(raw)
+ except Exception:
+ self._learned_priors = None
+
+ def _get_prior(self, cause_key):
+ """Return the learned prior for a cause, falling back to the hardcoded default."""
+ if self._learned_priors and cause_key in self._learned_priors:
+ return self._learned_priors[cause_key]
+ return _DEFAULT_PRIORS.get(cause_key, 0.05)
+
+ def _persist_crash(
+ self, error_type, error_msg, causes, symptoms, entropy, divergence
+ ):
+ """Store crash event in crash_history for future learning."""
+ if not self.database:
+ return
+ try:
+ top_cause = causes[0]["description"] if causes else "Unknown"
+ top_prob = causes[0]["probability"] if causes else 0
+ self.database.crash_history.insert_crash(
+ timestamp=time.time(),
+ error_type=error_type,
+ error_message=error_msg,
+ diagnosed_cause=top_cause,
+ symptoms=symptoms,
+ probability=top_prob,
+ entropy=entropy,
+ divergence=divergence,
+ )
+ self.database.crash_history.cleanup_old(max_entries=200)
+ except Exception:
+ pass
+
+ def _update_learned_weights(self):
+ """Bayesian weight update using Beta-Binomial conjugate model."""
+ if not self.database:
+ return
+ try:
+ freq_rows = self.database.crash_history.get_cause_frequencies(limit=50)
+ if not freq_rows:
+ return
+ total = sum(r["count"] for r in freq_rows)
+ if total < 3:
+ return
+
+ cause_counts = {r["diagnosed_cause"]: r["count"] for r in freq_rows}
+ weights = {}
+ for key, default_prior in _DEFAULT_PRIORS.items():
+ desc = self._cause_key_to_description(key)
+ count = cause_counts.get(desc, 0)
+ alpha = 1.0 + count
+ beta = 1.0 + (total - count)
+ posterior = alpha / (alpha + beta)
+ weights[key] = max(0.01, min(0.99, round(posterior, 4)))
+
+ self.database.config.set("diagnostic_weights", json.dumps(weights))
+ self._learned_priors = weights
+ except Exception:
+ pass
+
+ @staticmethod
+ def _cause_key_to_description(key):
+ """Map internal cause keys to their human-readable descriptions."""
+ mapping = {
+ "DB_SYNC_FAILURE": "In-Memory Database Sync Failure",
+ "DB_CORRUPTION": "SQLite Database Corruption",
+ "ASYNC_RACE": "Asynchronous Initialization Race Condition",
+ "OOM": "System Resource Exhaustion (OOM)",
+ "CONFIG_MISSING": "Missing Reticulum Configuration",
+ "RNS_IDENTITY_FAILURE": "Reticulum Identity Load Failure",
+ "LXMF_STORAGE_FAILURE": "LXMF Router Storage Failure",
+ "INTERFACE_OFFLINE": "Reticulum Interface Initialization Failure",
+ "UNSUPPORTED_PYTHON": "Unsupported Python Environment",
+ "LEGACY_SYSTEM_LIMITATION": "Legacy System Resource Limitation",
+ }
+ return mapping.get(key, key)
+
def handle_exception(self, exc_type, exc_value, exc_traceback):
"""Intercepts unhandled exceptions to provide a detailed diagnosis report."""
# Let keyboard interrupts pass through normally
@@ -113,6 +226,15 @@ class CrashRecovery:
out.write(f" [Manifold Curvature: {curvature:.2f}κ]\n")
out.write(" [Deterministic Manifold Constraints: V1,V4 Active]\n")
+ if self.log_handler:
+ log_ent = self.log_handler.current_log_entropy
+ err_rate = self.log_handler.current_error_rate
+ out.write(f" [Log Entropy (60s): {log_ent:.4f} bits]\n")
+ out.write(f" [Error Rate (60s): {err_rate:.2%}]\n")
+
+ if self._learned_priors:
+ out.write(" [Bayesian Priors: Learned from crash history]\n")
+
for cause in causes:
out.write(
f" - [{cause['probability']}% Probability] {cause['description']}\n",
@@ -146,6 +268,11 @@ class CrashRecovery:
out.write("=" * 70 + "\n\n")
out.flush()
+ # Persist crash and update weights (best-effort, never raise)
+ with contextlib.suppress(Exception):
+ self._persist_crash(error_type, error_msg, causes, {}, entropy, divergence)
+ self._update_learned_weights()
+
# Exit with error code
sys.exit(1)
@@ -157,10 +284,10 @@ class CrashRecovery:
error_msg = str(exc_value).lower()
error_type = exc_type.__name__.lower()
- # Define potential root causes with prior probabilities
+ # Define potential root causes with prior probabilities (learned or default)
potential_causes = {
"DB_SYNC_FAILURE": {
- "probability": 0.05,
+ "probability": self._get_prior("DB_SYNC_FAILURE"),
"description": "In-Memory Database Sync Failure",
"reasoning": "A background thread attempted to access an in-memory database that was not initialized in its local context.",
"suggestions": [
@@ -169,7 +296,7 @@ class CrashRecovery:
],
},
"DB_CORRUPTION": {
- "probability": 0.05,
+ "probability": self._get_prior("DB_CORRUPTION"),
"description": "SQLite Database Corruption",
"reasoning": "The database file on disk has become physically or logically corrupted.",
"suggestions": [
@@ -178,7 +305,7 @@ class CrashRecovery:
],
},
"ASYNC_RACE": {
- "probability": 0.10,
+ "probability": self._get_prior("ASYNC_RACE"),
"description": "Asynchronous Initialization Race Condition",
"reasoning": "A component tried to access the asyncio event loop before it was started.",
"suggestions": [
@@ -187,7 +314,7 @@ class CrashRecovery:
],
},
"OOM": {
- "probability": 0.02,
+ "probability": self._get_prior("OOM"),
"description": "System Resource Exhaustion (OOM)",
"reasoning": "Available system memory is extremely low, leading to allocation failures.",
"suggestions": [
@@ -196,7 +323,7 @@ class CrashRecovery:
],
},
"CONFIG_MISSING": {
- "probability": 0.01,
+ "probability": self._get_prior("CONFIG_MISSING"),
"description": "Missing Reticulum Configuration",
"reasoning": "The Reticulum Network Stack (RNS) could not find its configuration file.",
"suggestions": [
@@ -204,7 +331,7 @@ class CrashRecovery:
],
},
"RNS_IDENTITY_FAILURE": {
- "probability": 0.05,
+ "probability": self._get_prior("RNS_IDENTITY_FAILURE"),
"description": "Reticulum Identity Load Failure",
"reasoning": "The Reticulum identity file is missing, corrupt, or unreadable.",
"suggestions": [
@@ -213,7 +340,7 @@ class CrashRecovery:
],
},
"LXMF_STORAGE_FAILURE": {
- "probability": 0.05,
+ "probability": self._get_prior("LXMF_STORAGE_FAILURE"),
"description": "LXMF Router Storage Failure",
"reasoning": "The LXMF router could not access its message storage directory.",
"suggestions": [
@@ -222,7 +349,7 @@ class CrashRecovery:
],
},
"INTERFACE_OFFLINE": {
- "probability": 0.05,
+ "probability": self._get_prior("INTERFACE_OFFLINE"),
"description": "Reticulum Interface Initialization Failure",
"reasoning": "No active communication interfaces could be established.",
"suggestions": [
@@ -231,7 +358,7 @@ class CrashRecovery:
],
},
"UNSUPPORTED_PYTHON": {
- "probability": 0.05,
+ "probability": self._get_prior("UNSUPPORTED_PYTHON"),
"description": "Unsupported Python Environment",
"reasoning": "The application is running on an outdated or incompatible Python version.",
"suggestions": [
@@ -240,7 +367,7 @@ class CrashRecovery:
],
},
"LEGACY_SYSTEM_LIMITATION": {
- "probability": 0.05,
+ "probability": self._get_prior("LEGACY_SYSTEM_LIMITATION"),
"description": "Legacy System Resource Limitation",
"reasoning": "The host system lacks modern kernel features or resource allocation capabilities required for high-performance mesh networking.",
"suggestions": [
diff --git a/meshchatx/src/backend/recovery/health_monitor.py b/meshchatx/src/backend/recovery/health_monitor.py
new file mode 100644
index 00000000..748143ae
--- /dev/null
+++ b/meshchatx/src/backend/recovery/health_monitor.py
@@ -0,0 +1,156 @@
+"""Lightweight predictive health monitor.
+
+Runs as a daemon thread on a 5-minute interval, reading in-memory
+metrics (log entropy, error rate, memory) and emitting WebSocket
+warnings when anomalous trends are detected.
+
+No database queries are made in the monitor loop — all reads come
+from in-memory deques kept by PersistentLogHandler and psutil.
+"""
+
+import asyncio
+import collections
+import json
+import logging
+import threading
+
+import psutil
+
+_log = logging.getLogger("meshchatx.health_monitor")
+
+
+class HealthMonitor:
+ CHECK_INTERVAL = 300 # 5 minutes
+ ENTROPY_WINDOW = 6 # readings kept (~30 min at default interval)
+ ENTROPY_WARN_THRESHOLD = 1.5 # out of max ~2.32 for 5 log levels
+ ERROR_RATE_WARN = 0.3
+ MEMORY_WARN_MB = 100 # warn when available < 100 MB
+ CONSECUTIVE_NEEDED = 2 # consecutive bad readings before alert
+
+ def __init__(self, log_handler, app=None):
+ self.log_handler = log_handler
+ self.app = app
+ self._running = False
+ self._thread = None
+
+ self._entropy_history = collections.deque(maxlen=self.ENTROPY_WINDOW)
+ self._error_rate_history = collections.deque(maxlen=self.ENTROPY_WINDOW)
+ self._mem_available_history = collections.deque(maxlen=self.ENTROPY_WINDOW)
+
+ def start(self):
+ if self._running:
+ return
+ self._running = True
+ self._thread = threading.Thread(target=self._run_loop, daemon=True)
+ self._thread.start()
+ _log.info("HealthMonitor started (interval=%ds)", self.CHECK_INTERVAL)
+
+ def stop(self):
+ self._running = False
+
+ def _run_loop(self):
+ asyncio.run(self._async_loop())
+
+ async def _async_loop(self):
+ while self._running:
+ try:
+ self._check()
+ except Exception as exc:
+ _log.debug("HealthMonitor check error: %s", exc)
+ await asyncio.sleep(self.CHECK_INTERVAL)
+
+ def _check(self):
+ entropy = 0.0
+ error_rate = 0.0
+ if self.log_handler:
+ entropy = self.log_handler.current_log_entropy
+ error_rate = self.log_handler.current_error_rate
+
+ mem = psutil.virtual_memory()
+ available_mb = mem.available / (1024 * 1024)
+
+ self._entropy_history.append(entropy)
+ self._error_rate_history.append(error_rate)
+ self._mem_available_history.append(available_mb)
+
+ warnings = []
+
+ if self._detect_entropy_climb():
+ warnings.append(
+ {
+ "kind": "entropy_climbing",
+ "message": f"Log entropy rising: {entropy:.2f} bits (threshold {self.ENTROPY_WARN_THRESHOLD})",
+ "value": round(entropy, 4),
+ }
+ )
+
+ if self._consecutive_above(self._error_rate_history, self.ERROR_RATE_WARN):
+ warnings.append(
+ {
+ "kind": "error_rate_high",
+ "message": f"Error rate elevated: {error_rate:.0%}",
+ "value": round(error_rate, 4),
+ }
+ )
+
+ if self._consecutive_below(self._mem_available_history, self.MEMORY_WARN_MB):
+ warnings.append(
+ {
+ "kind": "memory_low",
+ "message": f"Available memory low: {available_mb:.0f} MB",
+ "value": round(available_mb, 1),
+ }
+ )
+
+ for w in warnings:
+ _log.warning("Health warning: %s", w["message"])
+ self._broadcast(w)
+
+ def _detect_entropy_climb(self):
+ """True if 3+ consecutive entropy readings are strictly increasing
+ AND the latest exceeds the warning threshold."""
+ h = self._entropy_history
+ if len(h) < 3:
+ return False
+ vals = list(h)
+ last_three = vals[-3:]
+ if last_three[-1] <= self.ENTROPY_WARN_THRESHOLD:
+ return False
+ return last_three[0] < last_three[1] < last_three[2]
+
+ @staticmethod
+ def _consecutive_above(deq, threshold):
+ if len(deq) < 2:
+ return False
+ vals = list(deq)
+ return vals[-1] > threshold and vals[-2] > threshold
+
+ @staticmethod
+ def _consecutive_below(deq, threshold):
+ if len(deq) < 2:
+ return False
+ vals = list(deq)
+ return vals[-1] < threshold and vals[-2] < threshold
+
+ def _broadcast(self, warning_data):
+ if not self.app:
+ return
+ try:
+ from meshchatx.src.backend.async_utils import AsyncUtils
+
+ AsyncUtils.run_async(
+ self.app.websocket_broadcast(
+ json.dumps({"type": "health_warning", "data": warning_data})
+ )
+ )
+ except Exception:
+ pass
+
+ @property
+ def latest_snapshot(self):
+ """Return latest metrics as a dict (useful for diagnostics/tests)."""
+ return {
+ "entropy": list(self._entropy_history),
+ "error_rate": list(self._error_rate_history),
+ "memory_mb": list(self._mem_available_history),
+ }
diff --git a/tests/backend/test_crash_history.py b/tests/backend/test_crash_history.py
new file mode 100644
index 00000000..db59ff7a
--- /dev/null
+++ b/tests/backend/test_crash_history.py
@@ -0,0 +1,247 @@
+import json
+import os
+import shutil
+import tempfile
+import time
+import unittest
+
+from meshchatx.src.backend.database import Database
+from meshchatx.src.backend.recovery.crash_recovery import CrashRecovery, _DEFAULT_PRIORS
+
+
+class TestCrashHistoryDAO(unittest.TestCase):
+ def setUp(self):
+ self.test_dir = tempfile.mkdtemp()
+ self.db_path = os.path.join(self.test_dir, "test_crash.db")
+ self.db = Database(self.db_path)
+ self.db.initialize()
+
+ def tearDown(self):
+ self.db.close_all()
+ shutil.rmtree(self.test_dir)
+
+ def test_insert_and_retrieve(self):
+ self.db.crash_history.insert_crash(
+ timestamp=time.time(),
+ error_type="ValueError",
+ error_message="test error",
+ diagnosed_cause="DB_CORRUPTION",
+ symptoms={"sqlite_in_msg": True},
+ probability=85,
+ entropy=1.234,
+ divergence=0.567,
+ )
+ rows = self.db.crash_history.get_recent_crashes(limit=10)
+ self.assertEqual(len(rows), 1)
+ self.assertEqual(rows[0]["error_type"], "ValueError")
+ self.assertEqual(rows[0]["diagnosed_cause"], "DB_CORRUPTION")
+ self.assertEqual(rows[0]["probability"], 85)
+
+ def test_get_cause_frequencies(self):
+ now = time.time()
+ for i in range(5):
+ self.db.crash_history.insert_crash(
+ timestamp=now + i,
+ error_type="E",
+ error_message="m",
+ diagnosed_cause="DB_CORRUPTION",
+ symptoms={},
+ probability=50,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ for i in range(3):
+ self.db.crash_history.insert_crash(
+ timestamp=now + 10 + i,
+ error_type="E",
+ error_message="m",
+ diagnosed_cause="OOM",
+ symptoms={},
+ probability=30,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ freqs = self.db.crash_history.get_cause_frequencies(limit=50)
+ freq_map = {r["diagnosed_cause"]: r["count"] for r in freqs}
+ self.assertEqual(freq_map["DB_CORRUPTION"], 5)
+ self.assertEqual(freq_map["OOM"], 3)
+
+ def test_cleanup_old(self):
+ now = time.time()
+ for i in range(20):
+ self.db.crash_history.insert_crash(
+ timestamp=now + i,
+ error_type="E",
+ error_message="m",
+ diagnosed_cause="X",
+ symptoms={},
+ probability=0,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ self.db.crash_history.cleanup_old(max_entries=5)
+ rows = self.db.crash_history.get_recent_crashes(limit=100)
+ self.assertEqual(len(rows), 5)
+
+ def test_insert_long_error_message_truncated(self):
+ long_msg = "x" * 1000
+ self.db.crash_history.insert_crash(
+ timestamp=time.time(),
+ error_type="E",
+ error_message=long_msg,
+ diagnosed_cause="X",
+ symptoms={},
+ probability=0,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ rows = self.db.crash_history.get_recent_crashes()
+ self.assertLessEqual(len(rows[0]["error_message"]), 500)
+
+ def test_symptoms_stored_as_json(self):
+ syms = {"sqlite_in_msg": True, "low_mem": False}
+ self.db.crash_history.insert_crash(
+ timestamp=time.time(),
+ error_type="E",
+ error_message="m",
+ diagnosed_cause="X",
+ symptoms=syms,
+ probability=0,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ rows = self.db.crash_history.get_recent_crashes()
+ loaded = json.loads(rows[0]["symptoms"])
+ self.assertEqual(loaded, syms)
+
+
+class TestBayesianWeightLearning(unittest.TestCase):
+ def setUp(self):
+ self.test_dir = tempfile.mkdtemp()
+ self.db_path = os.path.join(self.test_dir, "test_bayes.db")
+ self.db = Database(self.db_path)
+ self.db.initialize()
+ self.recovery = CrashRecovery(
+ storage_dir=self.test_dir,
+ database_path=self.db_path,
+ database=self.db,
+ )
+
+ def tearDown(self):
+ self.db.close_all()
+ shutil.rmtree(self.test_dir)
+
+ def test_get_prior_defaults(self):
+ for key, default_val in _DEFAULT_PRIORS.items():
+ self.assertEqual(self.recovery._get_prior(key), default_val)
+
+ def test_get_prior_learned(self):
+ self.recovery._learned_priors = {"DB_CORRUPTION": 0.42}
+ self.assertEqual(self.recovery._get_prior("DB_CORRUPTION"), 0.42)
+ self.assertEqual(self.recovery._get_prior("OOM"), _DEFAULT_PRIORS["OOM"])
+
+ def test_update_learned_weights_no_data(self):
+ self.recovery._update_learned_weights()
+ self.assertIsNone(self.recovery._learned_priors)
+
+ def test_update_learned_weights_with_history(self):
+ now = time.time()
+ desc = CrashRecovery._cause_key_to_description("DB_CORRUPTION")
+ for i in range(10):
+ self.db.crash_history.insert_crash(
+ timestamp=now + i,
+ error_type="E",
+ error_message="m",
+ diagnosed_cause=desc,
+ symptoms={},
+ probability=50,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ self.recovery._update_learned_weights()
+ self.assertIsNotNone(self.recovery._learned_priors)
+ self.assertGreater(self.recovery._learned_priors["DB_CORRUPTION"], 0.01)
+ self.assertLessEqual(self.recovery._learned_priors["DB_CORRUPTION"], 0.99)
+
+ def test_weights_clamped(self):
+ now = time.time()
+ desc = CrashRecovery._cause_key_to_description("DB_CORRUPTION")
+ for i in range(50):
+ self.db.crash_history.insert_crash(
+ timestamp=now + i,
+ error_type="E",
+ error_message="m",
+ diagnosed_cause=desc,
+ symptoms={},
+ probability=99,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ self.recovery._update_learned_weights()
+ for v in self.recovery._learned_priors.values():
+ self.assertGreaterEqual(v, 0.01)
+ self.assertLessEqual(v, 0.99)
+
+ def test_load_learned_priors_from_config(self):
+ weights = {"DB_CORRUPTION": 0.55, "OOM": 0.12}
+ self.db.config.set("diagnostic_weights", json.dumps(weights))
+ self.recovery._load_learned_priors()
+ self.assertEqual(self.recovery._learned_priors, weights)
+
+ def test_load_learned_priors_no_config(self):
+ self.recovery._load_learned_priors()
+ self.assertIsNone(self.recovery._learned_priors)
+
+ def test_set_database_loads_priors(self):
+ weights = {"DB_CORRUPTION": 0.33}
+ self.db.config.set("diagnostic_weights", json.dumps(weights))
+ new_recovery = CrashRecovery()
+ new_recovery.set_database(self.db)
+ self.assertEqual(new_recovery._learned_priors, weights)
+
+ def test_persist_crash(self):
+ causes = [
+ {"description": "Test Cause", "probability": 75},
+ ]
+ self.recovery._persist_crash("TypeError", "test msg", causes, {}, 1.5, 0.3)
+ rows = self.db.crash_history.get_recent_crashes()
+ self.assertEqual(len(rows), 1)
+ self.assertEqual(rows[0]["diagnosed_cause"], "Test Cause")
+ self.assertEqual(rows[0]["probability"], 75)
+
+ def test_persist_crash_no_db(self):
+ recovery_no_db = CrashRecovery()
+ recovery_no_db._persist_crash("E", "m", [], {}, 0.0, 0.0)
+
+ def test_cause_key_to_description_mapping(self):
+ for key in _DEFAULT_PRIORS:
+ desc = CrashRecovery._cause_key_to_description(key)
+ self.assertNotEqual(desc, key)
+ self.assertIsInstance(desc, str)
+
+ def test_cause_key_to_description_unknown(self):
+ self.assertEqual(
+ CrashRecovery._cause_key_to_description("UNKNOWN_KEY"),
+ "UNKNOWN_KEY",
+ )
+
+ def test_update_weights_needs_min_3_crashes(self):
+ now = time.time()
+ desc = CrashRecovery._cause_key_to_description("DB_CORRUPTION")
+ for i in range(2):
+ self.db.crash_history.insert_crash(
+ timestamp=now + i,
+ error_type="E",
+ error_message="m",
+ diagnosed_cause=desc,
+ symptoms={},
+ probability=50,
+ entropy=0.0,
+ divergence=0.0,
+ )
+ self.recovery._update_learned_weights()
+ self.assertIsNone(self.recovery._learned_priors)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/backend/test_health_monitor.py b/tests/backend/test_health_monitor.py
new file mode 100644
index 00000000..79925a08
--- /dev/null
+++ b/tests/backend/test_health_monitor.py
@@ -0,0 +1,167 @@
+import math
+import time
+import unittest
+from unittest.mock import MagicMock, patch
+
+from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler
+from meshchatx.src.backend.recovery.health_monitor import HealthMonitor
+
+
+class TestPersistentLogEntropy(unittest.TestCase):
+ """Tests for the log entropy and error rate properties."""
+
+ def setUp(self):
+ self.handler = PersistentLogHandler(capacity=5000, flush_interval=999)
+
+ def test_entropy_zero_when_empty(self):
+ self.assertEqual(self.handler.current_log_entropy, 0.0)
+
+ def test_error_rate_zero_when_empty(self):
+ self.assertEqual(self.handler.current_error_rate, 0.0)
+
+ def test_entropy_zero_single_level(self):
+ now = time.monotonic()
+ with self.handler.lock:
+ for _ in range(100):
+ self.handler._level_events.append((now, "INFO"))
+ self.assertAlmostEqual(self.handler.current_log_entropy, 0.0, places=5)
+
+ def test_entropy_max_uniform_distribution(self):
+ now = time.monotonic()
+ with self.handler.lock:
+ for level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
+ for _ in range(20):
+ self.handler._level_events.append((now, level))
+ expected = math.log2(5)
+ self.assertAlmostEqual(self.handler.current_log_entropy, expected, places=3)
+
+ def test_entropy_ignores_old_events(self):
+ old = time.monotonic() - 120.0
+ now = time.monotonic()
+ with self.handler.lock:
+ for level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
+ for _ in range(20):
+ self.handler._level_events.append((old, level))
+ for _ in range(50):
+ self.handler._level_events.append((now, "ERROR"))
+ self.assertAlmostEqual(self.handler.current_log_entropy, 0.0, places=5)
+
+ def test_error_rate_all_errors(self):
+ now = time.monotonic()
+ with self.handler.lock:
+ for _ in range(50):
+ self.handler._level_events.append((now, "ERROR"))
+ self.handler._error_events.append(now)
+ self.assertAlmostEqual(self.handler.current_error_rate, 1.0)
+
+ def test_error_rate_half(self):
+ now = time.monotonic()
+ with self.handler.lock:
+ for _ in range(50):
+ self.handler._level_events.append((now, "INFO"))
+ for _ in range(50):
+ self.handler._level_events.append((now, "ERROR"))
+ self.handler._error_events.append(now)
+ self.assertAlmostEqual(self.handler.current_error_rate, 0.5)
+
+
+class TestHealthMonitorDetection(unittest.TestCase):
+ """Tests for HealthMonitor detection logic (no real threads)."""
+
+ def setUp(self):
+ self.log_handler = MagicMock()
+ self.log_handler.current_log_entropy = 0.5
+ self.log_handler.current_error_rate = 0.0
+ self.monitor = HealthMonitor(log_handler=self.log_handler, app=None)
+
+ def test_no_warnings_normal_state(self):
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ mock_bc.assert_not_called()
+
+ def test_entropy_climb_detected(self):
+ self.monitor._entropy_history.extend([0.8, 1.2, 1.4])
+ self.log_handler.current_log_entropy = 1.6
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ mock_bc.assert_called_once()
+ args = mock_bc.call_args[0][0]
+ self.assertEqual(args["kind"], "entropy_climbing")
+
+ def test_entropy_not_climbing_when_below_threshold(self):
+ self.monitor._entropy_history.extend([0.2, 0.4, 0.6])
+ self.log_handler.current_log_entropy = 0.8
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ mock_bc.assert_not_called()
+
+ def test_error_rate_high_warning(self):
+ self.log_handler.current_error_rate = 0.5
+ self.monitor._error_rate_history.append(0.4)
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ calls = [c[0][0] for c in mock_bc.call_args_list]
+ error_warnings = [c for c in calls if c["kind"] == "error_rate_high"]
+ self.assertEqual(len(error_warnings), 1)
+
+ @patch("meshchatx.src.backend.recovery.health_monitor.psutil")
+ def test_memory_low_warning(self, mock_psutil):
+ mem_mock = MagicMock()
+ mem_mock.available = 50 * 1024 * 1024 # 50 MB
+ mock_psutil.virtual_memory.return_value = mem_mock
+ self.monitor._mem_available_history.append(80.0)
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ calls = [c[0][0] for c in mock_bc.call_args_list]
+ mem_warnings = [c for c in calls if c["kind"] == "memory_low"]
+ self.assertEqual(len(mem_warnings), 1)
+
+ def test_latest_snapshot_structure(self):
+ self.monitor._check()
+ snap = self.monitor.latest_snapshot
+ self.assertIn("entropy", snap)
+ self.assertIn("error_rate", snap)
+ self.assertIn("memory_mb", snap)
+ self.assertEqual(len(snap["entropy"]), 1)
+
+ def test_entropy_climb_needs_three_readings(self):
+ self.monitor._entropy_history.extend([1.2, 1.6])
+ self.log_handler.current_log_entropy = 1.8
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ calls = [c[0][0] for c in mock_bc.call_args_list]
+ entropy_warns = [c for c in calls if c["kind"] == "entropy_climbing"]
+ self.assertEqual(len(entropy_warns), 1)
+
+ def test_entropy_no_climb_when_decreasing(self):
+ self.monitor._entropy_history.extend([2.0, 1.8, 1.6])
+ self.log_handler.current_log_entropy = 1.4
+ with patch.object(self.monitor, "_broadcast") as mock_bc:
+ self.monitor._check()
+ calls = [c[0][0] for c in mock_bc.call_args_list]
+ entropy_warns = [c for c in calls if c["kind"] == "entropy_climbing"]
+ self.assertEqual(len(entropy_warns), 0)
+
+ def test_stop_prevents_further_checks(self):
+ self.monitor.stop()
+ self.assertFalse(self.monitor._running)
+
+
+class TestHealthMonitorEdgeCases(unittest.TestCase):
+ def test_check_with_no_log_handler(self):
+ monitor = HealthMonitor(log_handler=None, app=None)
+ monitor._check()
+ snap = monitor.latest_snapshot
+ self.assertEqual(snap["entropy"], [0.0])
+
+ def test_deque_fixed_size(self):
+ monitor = HealthMonitor(log_handler=MagicMock(), app=None)
+ monitor.log_handler.current_log_entropy = 1.0
+ monitor.log_handler.current_error_rate = 0.0
+ for _ in range(20):
+ monitor._check()
+ self.assertEqual(len(monitor._entropy_history), monitor.ENTROPY_WINDOW)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py
index 1f472c07..d51e6c71 100644
--- a/tests/backend/test_property_based.py
+++ b/tests/backend/test_property_based.py
@@ -1045,3 +1045,38 @@ class TestCrashRecoveryMathProperties:
for c in causes:
assert isinstance(c["probability"], int)
assert 0 <= c["probability"] <= 100
+
+ @given(
+ prior=st.floats(min_value=0.01, max_value=0.99, allow_nan=False),
+ count=st.integers(min_value=0, max_value=100),
+ total=st.integers(min_value=3, max_value=200),
+ )
+ @settings(derandomize=True, deadline=None, max_examples=100)
+ def test_bayesian_posterior_bounded(self, prior, count, total):
+ """Beta-Binomial posterior must stay in [0.01, 0.99]."""
+ import math as m
+
+ count = min(count, total)
+ alpha = 1.0 + count
+ beta = 1.0 + (total - count)
+ posterior = alpha / (alpha + beta)
+ clamped = max(0.01, min(0.99, round(posterior, 4)))
+ assert m.isfinite(clamped)
+ assert 0.01 <= clamped <= 0.99
+
+ @given(
+ counts=st.lists(
+ st.integers(min_value=0, max_value=50), min_size=1, max_size=10
+ ),
+ )
+ @settings(derandomize=True, deadline=None, max_examples=50)
+ def test_bayesian_posteriors_sum_reasonable(self, counts):
+ """Posteriors from any crash distribution should be valid probabilities."""
+ total = sum(counts) + 1 # ensure non-zero
+ posteriors = []
+ for c in counts:
+ alpha = 1.0 + c
+ beta = 1.0 + max(0, total - c)
+ posteriors.append(max(0.01, min(0.99, alpha / (alpha + beta))))
+ for p in posteriors:
+ assert 0.01 <= p <= 0.99
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────